/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

/* This file was modified by Jan Willem Boer to make it
 * suitable for use with the Compact Framework */

using System;
using System.Collections.Generic;
using System.Threading;
using Thrift.Collections;
using Thrift.Protocol;
using Thrift.Transport;

namespace Thrift.Server
{
    /// <summary>
    /// Server that uses C# threads (as opposed to the ThreadPool) when handling requests
    /// </summary>
    public class TThreadedServer : TServer
    {
        private const int DEFAULT_MAX_THREADS = 100;
        private volatile bool stop = false;
        private readonly int maxThreads;

        private Queue<TTransport> clientQueue;
        private THashSet<Thread> clientThreads;
        private object clientLock;
        private Thread workerThread;

        public TThreadedServer(TProcessor processor, TServerTransport serverTransport)
            : this(processor, serverTransport,
                 new TTransportFactory(), new TTransportFactory(),
                 new TBinaryProtocol.Factory(), new TBinaryProtocol.Factory(),
                 DEFAULT_MAX_THREADS, DefaultLogDelegate)
        {
        }

        public TThreadedServer(TProcessor processor, TServerTransport serverTransport, LogDelegate logDelegate)
            : this(processor, serverTransport,
                 new TTransportFactory(), new TTransportFactory(),
                 new TBinaryProtocol.Factory(), new TBinaryProtocol.Factory(),
                 DEFAULT_MAX_THREADS, logDelegate)
        {
        }


        public TThreadedServer(TProcessor processor,
                                 TServerTransport serverTransport,
                                 TTransportFactory transportFactory,
                                 TProtocolFactory protocolFactory)
            : this(processor, serverTransport,
                 transportFactory, transportFactory,
                 protocolFactory, protocolFactory,
                 DEFAULT_MAX_THREADS, DefaultLogDelegate)
        {
        }

        public TThreadedServer(TProcessor processor,
                                 TServerTransport serverTransport,
                                 TTransportFactory inputTransportFactory,
                                 TTransportFactory outputTransportFactory,
                                 TProtocolFactory inputProtocolFactory,
                                 TProtocolFactory outputProtocolFactory,
                                 int maxThreads, LogDelegate logDel)
            : base(processor, serverTransport, inputTransportFactory, outputTransportFactory,
                  inputProtocolFactory, outputProtocolFactory, logDel)
        {
            this.maxThreads = maxThreads;
            clientQueue = new Queue<TTransport>();
            clientLock = new object();
            clientThreads = new THashSet<Thread>();
        }

        /// <summary>
        /// Use new Thread for each new client connection. block until numConnections < maxTHreads
        /// </summary>
        public override void Serve()
        {
            try
            {
                //start worker thread
                workerThread = new Thread(new ThreadStart(Execute));
                workerThread.Start();
                serverTransport.Listen();
            }
            catch (TTransportException ttx)
            {
                logDelegate("Error, could not listen on ServerTransport: " + ttx);
                return;
            }

            while (!stop)
            {
                int failureCount = 0;
                try
                {
                    TTransport client = serverTransport.Accept();
                    lock (clientLock)
                    {
                        clientQueue.Enqueue(client);
                    }
                }
                catch (TTransportException ttx)
                {
                    if (stop)
                    {
                        logDelegate("TThreadPoolServer was shutting down, caught " + ttx);
                    }
                    else
                    {
                        ++failureCount;
                        logDelegate(ttx.ToString());
                    }

                }
            }

            if (stop)
            {
                try
                {
                    serverTransport.Close();
                }
                catch (TTransportException ttx)
                {
                    logDelegate("TServeTransport failed on close: " + ttx.Message);
                }
                stop = false;
            }
        }

        /// <summary>
        /// Loops on processing a client forever
        /// threadContext will be a TTransport instance
        /// </summary>
        /// <param name="threadContext"></param>
        private void Execute()
        {
            while (!stop)
            {
                Thread t = null;
                lock (clientLock)
                {
                    bool goAhead = false;
                    //don't dequeue if too many connections
                    goAhead = (clientThreads.Count < maxThreads);
                    goAhead = goAhead && (clientQueue.Count > 0);

                    if (goAhead)
                    {
                        TTransport client = clientQueue.Dequeue();
                        ClientWorker worker = new ClientWorker(client, this);

                        t = new Thread(new ThreadStart(worker.DoWork));
                        clientThreads.Add(t);
                    }
                }
                if (t == null)
                    Thread.Sleep(100);
                else
                    //start processing requests from client on new thread
                    t.Start();
            }
        }

        private class ClientWorker
        {
            private TTransport client;
            private TThreadedServer parent;
            public ClientWorker(TTransport client, TThreadedServer parent)
            {
                this.client = client;
                this.parent = parent;
            }

            public void DoWork()
            {
                TTransport inputTransport = null;
                TTransport outputTransport = null;
                TProtocol inputProtocol = null;
                TProtocol outputProtocol = null;
                try
                {
                    inputTransport = parent.inputTransportFactory.GetTransport(client);
                    outputTransport = parent.outputTransportFactory.GetTransport(client);
                    inputProtocol = parent.inputProtocolFactory.GetProtocol(inputTransport);
                    outputProtocol = parent.outputProtocolFactory.GetProtocol(outputTransport);
                    while (parent.processor.Process(inputProtocol, outputProtocol))
                    {
                        //keep processing requests until client disconnects
                    }
                }
                catch (TTransportException)
                {
                }
                catch (Exception x)
                {
                    parent.logDelegate("Error: " + x);
                }

                if (inputTransport != null)
                {
                    inputTransport.Close();
                }
                if (outputTransport != null)
                {
                    outputTransport.Close();
                }

                lock (parent.clientLock)
                {
                    parent.clientThreads.Remove(Thread.CurrentThread);
                }
                return;
            }
        }

        public override void Stop()
        {
            stop = true;
            serverTransport.Close();
            //clean up all the threads myself
            workerThread.Abort();
            foreach (Thread t in clientThreads)
            {
                t.Abort();
            }
        }
    }
}
